package com.wudl.flink.stream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author ：wudl
 * @date ：Created in 2021-11-28 18:41
 * @description： 使用侧输出流把一个流拆成多个流   flink 12 以后建议 多个流的输出用侧输入流
 * @modified By：
 * @version: 1.0
 */

public class Flink_SideOutputTwo {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> socketTextStream = env.socketTextStream("192.168.1.130", 9999);
        OutputTag<Integer> outputTag6 = new OutputTag<Integer>("number6-10") {
        };

        OutputTag<Integer> outputTag10 = new OutputTag<Integer>("number10") {
        };

        SingleOutputStreamOperator<Integer> process = socketTextStream.process(new ProcessFunction<String, Integer>() {
            @Override
            public void processElement(String s, Context cxt, Collector<Integer> out) throws Exception {
                int number = Integer.parseInt(s);
                if (number < 5) {
                    out.collect(number);
                } else if (number > 5 && number < 10) {
                    cxt.output(outputTag6, number);
                } else {
                    cxt.output(outputTag10, number);
                }

            }
        });
        process.print(" 主流-------->");
        process.getSideOutput(outputTag6).print("侧输入流--  大于5 小于10 的数据---");

        process.getSideOutput(outputTag10).print("侧输入流--  大于10 的数据---");

        env.execute();


    }

}
